package com.study.bigdata.spark.core.req

import com.study.bigdata.spark.core.summer.bean.UserVisitAction
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark03_Rep_PageFlow {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("Pageflow")
    val sc = new SparkContext(conf)
    val fileDatas = sc.textFile("data/user_visit_action.txt")

    val actionDatas = fileDatas.map(
      data => {
        val datas = data.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    actionDatas.cache()

    // 只统计关心的页面单跳转换率 过滤 (1,2),(2,3).....
    val okIds = List(1,2,3,4,5,6,7)
    val okFlowIds = okIds.zip(okIds.tail)

    // TODO 分母的计算
    val result = actionDatas.filter(
      action =>{
        okIds.init.contains(action.page_id.toInt)
      }
    ).map(
      action => {
        (action.page_id, 1)
      }
    ).reduceByKey(_ + _).collect().toMap
    // TODO 分子的计算
    // 将数据按照session分组
    val groupRDD = actionDatas.groupBy(_.session_id)

    // 将分组后的数据按照组内排序
    val mapRDD = groupRDD.mapValues(
      iter => {
        val actions = iter.toList.sortBy(_.action_time)
        val ids = actions.map(_.page_id.toInt)
        val flowIds = ids.zip(ids.tail)
        flowIds.filter(
          ids =>{
            okFlowIds.contains(ids)
          }
        )
      }
    )
    val mapRDD2 = mapRDD.map(_._2)
    val flatRDD = mapRDD2.flatMap(list => list)

    val reduceRDD= flatRDD.map((_,1)).reduceByKey(_+_)

    // TODO 单跳转换率的计算
    reduceRDD.foreach{
      case ((id1,id2),cnt) =>{
        println(s"页面[${id1}-${id2}]单跳转换率为："+(cnt.toDouble/result.getOrElse(id1,1)))
      }
    }


    sc.stop()

  }

}
